package com.kafka.study.producer;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.kafka.study.model.MessageDto;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

@Service
@Slf4j
public class KafkaProducer {
    @Value("${spring.kafka.template.time-topic}")
    private String timeTopic;

    @Resource(name = "timeKafkaTemplate")
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(MessageDto messageDto) throws ParseException, JsonProcessingException {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date date = sdf.parse(messageDto.getTime());
        long timestamp = date.getTime();
        ObjectMapper objectMapper = new ObjectMapper();
        String json = objectMapper.writeValueAsString(messageDto);
        Headers headers = new RecordHeaders();
        headers.add("traceId", "123456789".getBytes(StandardCharsets.UTF_8));
        ProducerRecord<String, String> record = new ProducerRecord<>(timeTopic, null, timestamp, messageDto.getMsgId(), json, headers);
        kafkaTemplate.send(record);
    }
}
